package com.ermao.mymes.support.pending;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.thread.ThreadUtil;
import com.google.common.base.Throwables;
import com.google.common.collect.Lists;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @author Ermao
 * Date: 2022/5/18 10:43
 */
@Data
@Slf4j
public abstract class AbstractLazyPending<T> {
	/**
	 * 子类构造方法必须初始化该参数
	 */
	protected PendingParam<T> pendingParam;

	/**
	 * 批量装载任务
	 */
	private List<T> tasks = new ArrayList<>();

	/**
	 * 上次执行的时间
	 */
	private Long lastHandleTime = System.currentTimeMillis();

	/**
	 * 单线程消费 阻塞队列的数据
	 */
	@PostConstruct
	public void initConsumePending() {
		// 当 TaskHandlerImpl 获取到一个实现类的对象时，此段代码会被执行，创建一个单例线程池：
		// 这个线程按照不断地从关联的阻塞队列中获取 csv 行信息
		ThreadUtil.newSingleExecutor().execute(() -> {
			while (true) {
				try {
					T obj = pendingParam.getQueue().poll(pendingParam.getTimeThreshold(), TimeUnit.MILLISECONDS);
					if (null != obj) {
						// 如果获得的信息不为 null，代表阻塞队列中存在要发送的消息
						// 添加到任务列表中
						tasks.add(obj);
					}

					// 处理条件：1. 数量超限 2. 时间超限
					if (CollUtil.isNotEmpty(tasks) && dataReady()) {
						List<T> taskRef = tasks;
						tasks = Lists.newArrayList();
						lastHandleTime = System.currentTimeMillis();

						// 具体执行逻辑
						pendingParam.getExecutorService().execute(() -> this.handle(taskRef));
					}
				} catch (Exception e) {
					log.error("Pending#initConsumePending failed:{}", Throwables.getStackTraceAsString(e));
				}
			}
		});
	}

	/**
	 * 1. 数量超限
	 * 2. 时间超限
	 * @return
	 */
	private boolean dataReady() {
		return tasks.size() >= pendingParam.getNumThreshold() ||
				(System.currentTimeMillis() - lastHandleTime >= pendingParam.getTimeThreshold());
	}

	/**
	 * 将元素放入阻塞队列中
	 *
	 * @param t
	 */
	public void pending(T t) {
		try {
			pendingParam.getQueue().put(t);
		} catch (InterruptedException e) {
			log.error("Pending#pending error:{}", Throwables.getStackTraceAsString(e));
		}
	}

	/**
	 * 消费阻塞队列元素时的方法
	 *
	 * @param t
	 */
	public void handle(List<T> t) {
		if (t.isEmpty()) {
			return;
		}
		try {
			doHandle(t);
		} catch (Exception e) {
			log.error("Pending#handle failed:{}", Throwables.getStackTraceAsString(e));
		}
	}

	/**
	 * 处理阻塞队列的元素 真正方法
	 *
	 * @param list
	 */
	public abstract void doHandle(List<T> list);
}
